#! /usr/bin/perl -w
package Runner;

use strict;
use Fcntl;
use IO::Handle;
use Config::ApacheFormat;
use POSIX qw(WNOHANG);
use Communicator;
use Constants;
use Message;
use Task;

our $VERSION = 1.00;

sub new{
	my ($class) = @_;
	my $self = {
		_id => undef,
		_host => undef,
		_db => undef,
		_dbuser => undef,
		_dbpasswd => undef,
		_identity => undef,
		_capability => undef,
		_communicator=>undef,
		_logdir => undef
	};
	bless $self;
	return $self;
}

sub init{
	my ($self) = @_;
	$self->{_communicator} = new Communicator();

	#read configuration from conf file
	my $config = Config::ApacheFormat->new();
	$config->read("tabot.conf");

	my $block_runner = $config->block("runner"=>"general");
	my $block_server = $config->block("server"=>"database");

   $self->{_host} = $block_server->get("HOST");
	$self->{_db} = $block_server->get("DB_NAME");
	$self->{_dbuser} = $block_server->get("DB_USERNAME");
	$self->{_dbpasswd} = $block_server->get("DB_PASSWORD");

	#store the general information about this runner
   $self->{_identity} = $block_runner->get("IDENTITY");
	$self->{_capability} = $block_runner->get("CAPABILITY");
	$self->{_logdir} = $block_runner->get("LOGDIR");

	return 0;
}

sub run{
   my ($self) = @_;
   my $buf = undef;

   #setup the pipe between processes to communicate
   our ($PARENT_RDR,  $CHILD_WTR);
   my ($CHILD_RDR,  $PARENT_WTR);
   
   pipe($PARENT_RDR, $CHILD_WTR);
   pipe($CHILD_RDR,  $PARENT_WTR);

   $CHILD_WTR->autoflush(1);
   $PARENT_WTR->autoflush(1);

   my $flags = fcntl($PARENT_RDR, F_GETFL, 0) or die "Can't get flags for the pipe: $! \n";
   $flags = fcntl($PARENT_RDR, F_SETFL, $flags | O_NONBLOCK) or die "Can't set the flags for the pipe: $! \n";
   $flags = fcntl($CHILD_RDR, F_GETFL, 0) or die "Can't get flags for the pipe: $! \n";
   $flags = fcntl($CHILD_RDR, F_SETFL, $flags | O_NONBLOCK) or die "Can't set the flags for the pipe: $! \n";

	#create child process to execute the task
   my $pid = 0;

   #fork a child process to handle the communication between host and database
   defined($pid = fork) or die "Can't fork for init: $! \n";
   if($pid == 0)
   {
      close $PARENT_RDR;
      close $PARENT_WTR;

		#register runner in database
		$self->{_communicator}->init($self->{_host}, $self->{_db}, $self->{_dbuser}, 
			$self->{_dbpasswd}, $self->{_identity}, $self->{_capability});
		$self->{_id} = $self->{_communicator}->register();
      
      while(1)
      {
	   	$self->handle_interrupts();
      
      	my $tasks = $self->{_communicator}->fetch_task_executable();

      	if(defined($tasks))
      	{
      	   print "start a task...\n";
         	print $CHILD_WTR $COMMAND_TAG."$tasks->{id};$tasks->{logfile};$tasks->{command}\n";
         	$self->{_communicator}->update_task_stage($tasks->{id}, $TASK_RUNNING);
				#comsume capability, reduce the piece of capability this runner
         	$self->{_communicator}->consume_capability($tasks->{capabilities});
      	}
      	   
        	while(<$CHILD_RDR>) {
	      	my $message = $self->{_communicator}->parse_message($_);

	        	if($message->get_type()==0) {
		     		if($message->get_executable() == $TASK_RUNNING) {
		        		$self->{_communicator}->update_task_status($message->get_target_id, $message->get_parameter());
		         } elsif($message->get_executable() == $TASK_SCHEDULED) {
		         	$self->{_communicator}->update_task_pid($message->get_target_id, $message->get_parameter());
		         } elsif($message->get_executable() == $TASK_END) {
			      	$self->{_communicator}->recycle_capability("svn:1,make:1");
		           	$self->{_communicator}->update_task_stage($message->get_target_id, $message->get_executable());
		         }
	         }
        	}
			print "runner sleeps 5 seconds ...\n";
        	sleep 5;
      }
      close $CHILD_WTR;
      close $CHILD_RDR;
      exit;
   } else {
      close $CHILD_RDR;
      my %task_pids;
      while(1) {
         if((my $kidpid = waitpid(-1, WNOHANG))>0) {
				print "Child $kidpid ends\n";
            print $PARENT_WTR $STATUS_TAG."$task_pids{$kidpid}:$TASK_END:Test ended\n";
         }   

         while(($buf = <$PARENT_RDR>)) {

            if(defined($buf) && !($buf eq "")) {
               print "****: $buf\n"; 
               chomp($buf);

               my $message = $self->{_communicator}->parse_message($buf);

               if($message->get_type() == $MSG_COMMAND_TYPE) {
               	defined ($pid = fork) or die "Can't fork: $! \n";
                  if($pid==0){
                  	close $PARENT_RDR;
                     close $PARENT_WTR;

                     my  $task = new Task();
                     $task->init($message->get_target_id(),*$CHILD_WTR, 
								$self->{_logdir}."/".$message->get_target_id()."_".$message->get_logfile(),
								$message->get_executable(), $message->get_parameter());
                     $task->exec();
                        
                     close $CHILD_WTR;
                     exit;      
                  } else {
			 				my $tid = $message->get_target_id;
							$task_pids{$pid} = $tid;
                    	print $PARENT_WTR $STATUS_TAG."$tid:$TASK_SCHEDULED:$pid\n";
			  			}
               } elsif($message->get_type() == $MSG_STATUS_TYPE) {
                  print $PARENT_WTR "$buf\n";
               } else {
                  print "incorrect message received!\n";
               }
            }
         }      
         sleep 5;
      }
   }
}

sub destory {
	my ($self) = @_;
	$self->{_communicator}->destory;
}

sub handle_interrupts {
	my ($self) =  @_;

	my $hashref = $self->{_communicator}->fetch_task_interrupts();
	foreach my $task_id (sort(keys %$hashref)) {  	
   	my $interrupt = $hashref->{$task_id}->{"interrupt"};
		$self->stop_task($task_id, $interrupt);
	}	
}

sub stop_task {
	my ($self, $tid, $interrupt) = @_;

	my $pid = $self->{_communicator}->get_pid_by_tid($tid);
	return unless defined($pid);

	kill $interrupt, $pid;
	$self->{_communicator}->update_task_stage_and_interrupt($tid, $TASK_CANCELLED, 0);
}

